{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "pd.set_option('display.max_colwidth', 1000)\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"ctr_avazu\" + os.sep\n",
    "\n",
    "SCHEMA_STRING\\\n",
    "    = \"id string, click string, dt string, C1 string, banner_pos int, site_id string, site_domain string, \"\\\n",
    "    + \"site_category string, app_id string, app_domain string, app_category string, device_id string, \"\\\n",
    "    + \"device_ip string, device_model string, device_type string, device_conn_type string, C14 int, C15 int, \"\\\n",
    "    + \"C16 int, C17 int, C18 int, C19 int, C20 int, C21 int\"\n",
    "\n",
    "CATEGORY_COL_NAMES = [\n",
    "    \"C1\", \"banner_pos\", \"site_category\", \"app_domain\",\n",
    "    \"app_category\", \"device_type\", \"device_conn_type\",\n",
    "    \"site_id\", \"site_domain\", \"device_id\", \"device_model\"\n",
    "]\n",
    "\n",
    "NUMERICAL_COL_NAMES = [\"C14\", \"C15\", \"C16\", \"C17\", \"C18\", \"C19\", \"C20\", \"C21\"]\n",
    "\n",
    "FEATURE_MODEL_FILE = \"feature_model.ak\"\n",
    "INIT_MODEL_FILE = \"init_model.ak\"\n",
    "\n",
    "LABEL_COL_NAME = \"click\"\n",
    "VEC_COL_NAME = \"vec\"\n",
    "PREDICTION_COL_NAME = \"pred\"\n",
    "PRED_DETAIL_COL_NAME = \"pred_info\"\n",
    "\n",
    "NUM_HASH_FEATURES = 30000\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "TextSourceBatchOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-small.csv\")\\\n",
    "    .firstN(10)\\\n",
    "    .print()\n",
    "\n",
    "trainBatchData = CsvSourceBatchOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-small.csv\")\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "trainBatchData.firstN(10).print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "trainBatchData = CsvSourceBatchOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-small.csv\")\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "feature_pipeline = Pipeline()\\\n",
    "    .add(\n",
    "        StandardScaler()\\\n",
    "            .setSelectedCols(NUMERICAL_COL_NAMES)\n",
    "    )\\\n",
    "    .add(\n",
    "        FeatureHasher()\\\n",
    "            .setSelectedCols(CATEGORY_COL_NAMES + NUMERICAL_COL_NAMES)\\\n",
    "            .setCategoricalCols(CATEGORY_COL_NAMES)\\\n",
    "            .setOutputCol(VEC_COL_NAME)\\\n",
    "            .setNumFeatures(NUM_HASH_FEATURES)\n",
    "    );\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + FEATURE_MODEL_FILE)) :\n",
    "    feature_pipeline\\\n",
    "        .fit(trainBatchData)\\\n",
    "        .save(DATA_DIR + FEATURE_MODEL_FILE)\n",
    "    BatchOperator.execute()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_4\n",
    "feature_pipelineModel = PipelineModel.load(DATA_DIR + FEATURE_MODEL_FILE)\n",
    "\n",
    "data = CsvSourceStreamOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-ctr-train-8M.csv\")\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + INIT_MODEL_FILE)) :\n",
    "    trainBatchData = CsvSourceBatchOp()\\\n",
    "        .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                     + \"data-files/avazu-small.csv\")\\\n",
    "        .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "    lr = LogisticRegressionTrainBatchOp()\\\n",
    "        .setVectorCol(VEC_COL_NAME)\\\n",
    "        .setLabelCol(LABEL_COL_NAME)\\\n",
    "        .setWithIntercept(True)\\\n",
    "        .setMaxIter(10);\n",
    "\n",
    "    feature_pipelineModel\\\n",
    "    .transform(trainBatchData)\\\n",
    "    .link(lr)\\\n",
    "    .link(\n",
    "        AkSinkBatchOp().setFilePath(DATA_DIR + INIT_MODEL_FILE)\n",
    "    );\n",
    "    BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5 \n",
    "feature_pipelineModel = PipelineModel.load(DATA_DIR + FEATURE_MODEL_FILE);\n",
    "\n",
    "initModel = AkSourceBatchOp().setFilePath(DATA_DIR + INIT_MODEL_FILE);\n",
    "\n",
    "data = CsvSourceStreamOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-ctr-train-8M.csv\")\\\n",
    "    .setSchemaStr(SCHEMA_STRING)\\\n",
    "    .setIgnoreFirstLine(True)\n",
    "\n",
    "spliter = SplitStreamOp().setFraction(0.5).linkFrom(data);\n",
    "train_stream_data = feature_pipelineModel.transform(spliter);\n",
    "test_stream_data = feature_pipelineModel.transform(spliter.getSideOutput(0));\n",
    "\n",
    "model = FtrlTrainStreamOp(initModel)\\\n",
    "    .setVectorCol(VEC_COL_NAME)\\\n",
    "    .setLabelCol(LABEL_COL_NAME)\\\n",
    "    .setWithIntercept(True)\\\n",
    "    .setAlpha(0.1)\\\n",
    "    .setBeta(0.1)\\\n",
    "    .setL1(0.01)\\\n",
    "    .setL2(0.01)\\\n",
    "    .setTimeInterval(10)\\\n",
    "    .setVectorSize(NUM_HASH_FEATURES)\\\n",
    "    .linkFrom(train_stream_data);\n",
    "\n",
    "predResult = FtrlPredictStreamOp(initModel)\\\n",
    "    .setVectorCol(VEC_COL_NAME)\\\n",
    "    .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "    .setReservedCols([LABEL_COL_NAME])\\\n",
    "    .setPredictionDetailCol(PRED_DETAIL_COL_NAME)\\\n",
    "    .linkFrom(model, test_stream_data);\n",
    "\n",
    "# predResult\\\n",
    "#     .sample(0.0001)\\\n",
    "#     .select(\"'Pred Sample' AS out_type, *\")\\\n",
    "#     .print();\n",
    "\n",
    "predResult.print(key=\"predResult\", refreshInterval = 30, maxLimit=20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predResult\\\n",
    "    .link(\n",
    "        EvalBinaryClassStreamOp()\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .setPredictionDetailCol(PRED_DETAIL_COL_NAME)\\\n",
    "            .setTimeInterval(10)\n",
    "    )\\\n",
    "    .link(\n",
    "        JsonValueStreamOp()\\\n",
    "            .setSelectedCol(\"Data\")\\\n",
    "            .setReservedCols([\"Statistics\"])\\\n",
    "            .setOutputCols([\"Accuracy\", \"AUC\", \"ConfusionMatrix\"])\\\n",
    "            .setJsonPath([\"$.Accuracy\", \"$.AUC\", \"$.ConfusionMatrix\"])\n",
    "    )\\\n",
    "    .print(key=\"evaluation\", refreshInterval = 30, maxLimit=20)\n",
    "# .select(\"'Eval Metric' AS out_type, *\")\\\n",
    "#     .print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_6\n",
    "data = CsvSourceStreamOp()\\\n",
    "    .setFilePath(\"http://alink-release.oss-cn-beijing.aliyuncs.com/\"\n",
    "                 + \"data-files/avazu-ctr-train-8M.csv\")\\\n",
    "    .setSchemaStr(SCHEMA_STRING)\\\n",
    "    .setIgnoreFirstLine(True);\n",
    "\n",
    "feature_pipelineModel = PipelineModel.load(DATA_DIR + FEATURE_MODEL_FILE);\n",
    "\n",
    "spliter = SplitStreamOp().setFraction(0.5).linkFrom(data);\n",
    "train_stream_data = feature_pipelineModel.transform(spliter);\n",
    "test_stream_data = feature_pipelineModel.transform(spliter.getSideOutput(0));\n",
    "\n",
    "initModel = AkSourceBatchOp().setFilePath(DATA_DIR + INIT_MODEL_FILE);\n",
    "\n",
    "model = FtrlTrainStreamOp(initModel)\\\n",
    "    .setVectorCol(VEC_COL_NAME)\\\n",
    "    .setLabelCol(LABEL_COL_NAME)\\\n",
    "    .setWithIntercept(True)\\\n",
    "    .setAlpha(0.1)\\\n",
    "    .setBeta(0.1)\\\n",
    "    .setL1(0.01)\\\n",
    "    .setL2(0.01)\\\n",
    "    .setTimeInterval(10)\\\n",
    "    .setVectorSize(NUM_HASH_FEATURES)\\\n",
    "    .linkFrom(train_stream_data);\n",
    "\n",
    "model_filter = FtrlModelFilterStreamOp()\\\n",
    "    .setPositiveLabelValueString(\"1\")\\\n",
    "    .setVectorCol(VEC_COL_NAME)\\\n",
    "    .setLabelCol(LABEL_COL_NAME)\\\n",
    "    .setAccuracyThreshold(0.83)\\\n",
    "    .setAucThreshold(0.71)\\\n",
    "    .linkFrom(model, train_stream_data);\n",
    "\n",
    "model_filter\\\n",
    "    .select(\"'Model' AS out_type, *\")\\\n",
    "    .print();\n",
    "\n",
    "predResult = FtrlPredictStreamOp(initModel)\\\n",
    "    .setVectorCol(VEC_COL_NAME)\\\n",
    "    .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "    .setReservedCols([LABEL_COL_NAME])\\\n",
    "    .setPredictionDetailCol(PRED_DETAIL_COL_NAME)\\\n",
    "    .linkFrom(model_filter, test_stream_data);\n",
    "\n",
    "predResult\\\n",
    "    .sample(0.0001)\\\n",
    "    .select(\"'Pred Sample' AS out_type, *\")\\\n",
    "    .print();\n",
    "\n",
    "predResult\\\n",
    "    .link(\n",
    "        EvalBinaryClassStreamOp()\\\n",
    "            .setPositiveLabelValueString(\"1\")\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .setPredictionDetailCol(PRED_DETAIL_COL_NAME)\\\n",
    "            .setTimeInterval(10)\n",
    "    )\\\n",
    "    .link(\n",
    "        JsonValueStreamOp()\\\n",
    "            .setSelectedCol(\"Data\")\\\n",
    "            .setReservedCols([\"Statistics\"])\\\n",
    "            .setOutputCols([\"Accuracy\", \"AUC\", \"ConfusionMatrix\"])\\\n",
    "            .setJsonPath([\"$.Accuracy\", \"$.AUC\", \"$.ConfusionMatrix\"])\n",
    "    )\\\n",
    "    .select(\"'Eval Metric' AS out_type, *\")\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
